package me.tuyou.wordcount;

import me.tuyou.common.WordStreamGenerator;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

public class SQLStreamWordCountTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<String> dataStream = env.addSource(new WordStreamGenerator());
        Table table = tableEnv.fromDataStream(dataStream, $("word"));

        String sql = "select * from " + table + " where word like '%t%'";
        System.out.println("sql:" + sql);
        Table result = tableEnv.sqlQuery(sql);

        tableEnv.toAppendStream(result, Row.class).print();
        env.execute();
    }
}
